package com.example.window;

import com.example.beans.Event;
import com.example.source.ClickSource;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.HashSet;

/**
 *
 *  * @projectName myflinkstu
 *  * @title     SlidingEventTimeWindow_AggregateExample
 *  * @package    com.example.window
 *  * @description    滑动事件时间+聚合函数
 *  * @author hjx
 *  * @date   2022-3-28 15:25
 *  * @version V1.0.0
 *  * @copyright 2022 ty
 *
 */
public class SlidingEventTimeWindow_AggregateExample {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> dataStreamSource = env.addSource(new ClickSource(1000L));

        // 获取事件时间戳, 添加水位线, 有序流
        SingleOutputStreamOperator<Event> streamWithWaterMark = dataStreamSource.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        })
        );

        SingleOutputStreamOperator<Double> result = streamWithWaterMark.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event value) throws Exception {
                return "true";
            }
        }).window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(2)))
                .aggregate(new MyAvg());


        dataStreamSource.print("dataStreamSource ");

        result.print("SlidingEventTimeWindow_AggregateExample ");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }


     // AggregateFunction<IN, ACC, OUT>
    private static class MyAvg implements AggregateFunction<Event, Tuple2<HashSet<String>, Long>, Double> {

        @Override
        public Tuple2<HashSet<String>, Long> createAccumulator() {
            // 创建累加器
            return Tuple2.of(new HashSet<String>(), 0L);
        }

        @Override
        public Tuple2<HashSet<String>, Long> add(Event value, Tuple2<HashSet<String>, Long> accumulator) {
            // 属于本窗口的数据来一条累加一次，并返回累加器
            accumulator.f0.add(value.user);
             // user.size -> UV , accumulator.f1 + 1L -> PV
            return Tuple2.of(accumulator.f0, accumulator.f1 + 1L);
        }

        @Override
        public Double getResult(Tuple2<HashSet<String>, Long> accumulator) {
            // 窗口闭合时，增量聚合结束，将计算结果发送到下游
            return (double) accumulator.f1 / accumulator.f0.size();
        }

        // merge 方法多用于 会话窗口使用
        @Override
        public Tuple2<HashSet<String>, Long> merge(Tuple2<HashSet<String>, Long> a, Tuple2<HashSet<String>, Long> b) {
            return null;
        }
    }
}
